package com.sunlands.deskmate.message.handler.mq;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.sunlands.deskmate.config.RabbitMqConfig;
import com.sunlands.deskmate.message.entity.Messages;
import com.sunlands.deskmate.message.handler.dispatcher.DispatcherManager;
import com.sunlands.deskmate.message.handler.storager.interceptor.MessagePreInterceptorChain;
import com.sunlands.deskmate.utils.DateUtil;
import com.sunlands.deskmate.utils.IdGenerator;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import static com.sunlands.deskmate.config.RabbitMqConfig.EXCHANGE_ERROR_MESSAGE;
import static com.sunlands.deskmate.config.RabbitMqConfig.QUEUE_STORAGE;
import static com.sunlands.deskmate.config.RedisConfig.PREFIX;


/**
 * STORAGE_QUEUE -> unsyncStorager -> LOOKUP_QUEUE
 * <p>
 * 从gateway通过storage队列发来的消息, 处理之后进入lookup队列
 *
 * @author anjl
 * @author anjunling@sunlands.com
 */
@Slf4j
@Component
public class StorageMessageWorker extends AbstractInboundMessageWorker {

    public static final String STORAGE = "storage";

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private RabbitMqConfig rabbitMqConfig;

    @Autowired
    private MessagePreInterceptorChain messagePreInterceptorChain;

    @Autowired
    private DispatcherManager dispatcherFactory;
    /**
     * 等待锁时间
     */
    public static final int waitTimeInSeconds = 3;
    /**
     * 锁过期时间
     */
    public static final int leaseTimeInSeconds = 6;

    /**
     * 从storage queue里获取消息,
     * 记录unsync消息
     * 发送到lookup queue
     *
     * @param mqMessage mqMessage
     * @param channel   channel
     */
    @RabbitListener(queues = QUEUE_STORAGE + "_" + "${spring.profiles.active}")
    public void run(Message mqMessage, Channel channel) throws IOException {
        String message = new String(mqMessage.getBody());

        log.info("保存消息到未读库 storage queue {}", message);
        RLock fairLock = null;
        try {
            if (StringUtils.isEmpty(message)) {
                log.warn("messages is empty");
                handle(true, mqMessage, channel);
                return;
            }
            Messages msgEntity = JSONObject.parseObject(message, Messages.class);
            String checkDesc = checkMsg(msgEntity);
            if (checkDesc != null) {
                log.warn("{} messages invalid {}", checkDesc, message);
                handle(true, mqMessage, channel);
                return;
            }

            String requestId = msgEntity.getRequestId();
            String toId = msgEntity.getToId();
            String type = msgEntity.getType();

            String lockKey = lockKey(type, toId);
            fairLock = redissonClient.getFairLock(lockKey);
            if (fairLock.tryLock(waitTimeInSeconds, leaseTimeInSeconds, TimeUnit.SECONDS)) {
                //TODO暂时关闭验证消息幂等
                boolean valid;
//                valid = idempotent.idempotentCheck(STORAGE, requestId);
//                if (!valid) {
//                    log.warn("messages is duplicate {}", message);
//                    handle(true, mqMessage, channel);
//                    return;
//                }

                //存储未同步消息
                long msgId = msgEntity.getAckId();

                if (msgId <= 0) {
                    msgId = IdGenerator.id();
                    msgEntity.setAckId(msgId);
                    log.info("generate msgId={}, msgEntity={}", msgId, msgEntity);
                }

                String createDateTime = msgEntity.getCreateTime();
                if (Objects.isNull(createDateTime)) {
                    msgEntity.setCreateTime(DateUtil.getStringDate());
                }

                //所有检查或者预处理通过后才能发送消息
                if(messagePreInterceptorChain.interceptor(msgEntity)){
                    dispatcherFactory.getDispatcher(msgEntity.getType()).dispatcher(msgEntity);
                }
                handle(true, mqMessage, channel);
            } else {
                log.error("getLock fail ,message:{}", message);
                //重试
                handle(false, mqMessage, channel);
            }
        } catch (IOException e) {
            log.warn("storage IOException  messages={}", message);
            log.error("", e);
            rabbitTemplate.convertAndSend(EXCHANGE_ERROR_MESSAGE, rabbitMqConfig.getEnv(), message);
            handle(true, mqMessage, channel);
        } catch (Exception e) {
            log.error("storage Exception messages = {}, e = {}", message, e);
            log.error("", e);
            rabbitTemplate.convertAndSend(EXCHANGE_ERROR_MESSAGE, rabbitMqConfig.getEnv(), message);
            handle(true, mqMessage, channel);
        } finally {
            if (Objects.nonNull(fairLock)) {
                try {
                    fairLock.unlock();
                } catch (Exception ignored) {
                    log.error("unlock error ", ignored);
                }
            }
        }
    }


    private String lockKey(String type, String toId) {
        return PREFIX + ":storage:type_" + type + "_" + toId;
    }

}
